package rxgo

import (
	"context"
	"errors"
	"reflect"
	"sync"
	"time"
)

// ErrBoundOut indicates out of bound
var ErrBoundOut = errors.New("Out Of Bound")

// filter node implementation of streamOperator
type filterOperator struct {
	opFunc func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool)
}

func (ftop filterOperator) op(ctx context.Context, o *Observable) {
	// must hold defintion of flow resourcs here, such as chan etc., that is allocated when connected
	// this resurces may be changed when operation routine is running.
	in := o.pred.outflow
	out := o.outflow
	//fmt.Println(o.name, "operator in/out chan ", in, out)
	var wg sync.WaitGroup

	tspan := o.debounce
	var _out []interface{}

	go func() {
		end := false
		//flag 用来检测是否出现相同的数据
		flag := make(map[interface{}]bool)

		// 获取开始的时间
		start := time.Now()
		sampleStart := time.Now()

		for x := range in {
			_start := time.Since(start)
			_sample := time.Since(sampleStart)
			start = time.Now()

			if end {
				continue
			}

			if o.sample > 0 && _sample < o.sample {
				continue
			}

			if tspan > time.Duration(0) && _start < tspan {
				continue
			}
			// can not pass a interface as parameter (pointer) to gorountion for it may change its value outside!
			xv := reflect.ValueOf(x)
			//fmt.Println(flag[xv.Interface()])
			// send an error to stream if the flip not accept error
			if e, ok := x.(error); ok && !o.flip_accept_error {
				o.sendToFlow(ctx, e, out)
				continue
			}

			o.mu.Lock()
			_out = append(_out, x)
			o.mu.Unlock()

			if o.elementAt > 0 {
				continue
			}

			if o.take != 0 {
				continue
			}

			if o.takeLast != 0 {
				continue
			}

			if o.skip != 0 {
				continue
			}

			if o.skipLast != 0 {
				continue
			}

			if o.first {
				continue
			}

			if o.last {
				continue
			}

			//去重
			if o.distinct && flag[xv.Interface()] {
				continue
			}
			o.mu.Lock()
			flag[xv.Interface()] = true
			o.mu.Unlock()

			// scheduler
			switch threading := o.threading; threading {
			case ThreadingDefault:
				if o.sample > 0 {
					sampleStart = sampleStart.Add(o.sample)
				}
				if ftop.opFunc(ctx, o, xv, out) {
					end = true
				}
			case ThreadingIO:
				fallthrough
			case ThreadingComputing:
				wg.Add(1)
				if o.sample > 0 {
					sampleStart.Add(o.sample)
				}
				go func() {
					defer wg.Done()
					if ftop.opFunc(ctx, o, xv, out) {
						end = true
					}
				}()
			default:
			}
		}

		if o.elementAt != 0 {
			if o.elementAt < 0 || o.elementAt > len(_out) {
				o.sendToFlow(ctx, ErrBoundOut, out)
			} else {
				xv := reflect.ValueOf(_out[o.elementAt-1])
				ftop.opFunc(ctx, o, xv, out)
			}
		}

		if o.first && len(_out) > 0 {
			wg.Add(1)
			go func() {
				defer wg.Done()
				xv := reflect.ValueOf(_out[0])
				ftop.opFunc(ctx, o, xv, out)
			}()
		}

		if o.last && len(_out) > 0 {
			wg.Add(1)
			go func() {
				defer wg.Done()
				xv := reflect.ValueOf(_out[len(_out)-1])
				ftop.opFunc(ctx, o, xv, out)
			}()
		}

		if o.skip != 0 {
			wg.Add(1)
			go func() {
				defer wg.Done()
				if len(_out) < o.skip {
					o.sendToFlow(ctx, ErrBoundOut, out)
				} else {
					for i := o.skip; i < len(_out); i++ {
						xv := reflect.ValueOf(_out[i])
						ftop.opFunc(ctx, o, xv, out)
					}
				}
			}()
		}

		if o.skipLast != 0 {
			wg.Add(1)
			go func() {
				defer wg.Done()
				if len(_out) < o.skipLast {
					o.sendToFlow(ctx, ErrBoundOut, out)
				} else {
					for i := 0; i < len(_out)-o.skipLast; i++ {
						xv := reflect.ValueOf(_out[i])
						ftop.opFunc(ctx, o, xv, out)
					}
				}
			}()
		}

		if o.take != 0 {
			wg.Add(1)
			go func() {
				defer wg.Done()
				if len(_out) < o.take {
					o.sendToFlow(ctx, ErrBoundOut, out)
				} else {
					for i := 0; i < o.take; i++ {
						xv := reflect.ValueOf(_out[i])
						ftop.opFunc(ctx, o, xv, out)
					}
				}
			}()
		}

		if o.takeLast != 0 {
			wg.Add(1)
			go func() {
				defer wg.Done()
				if len(_out) < o.takeLast {
					o.sendToFlow(ctx, ErrBoundOut, out)
				} else {
					for i := 0; i < o.takeLast; i++ {
						xv := reflect.ValueOf(_out[len(_out)-i-1])
						ftop.opFunc(ctx, o, xv, out)
					}
				}
			}()
		}

		wg.Wait()
		o.closeFlow(out)
	}()
}

func (parent *Observable) newFilterObservable(name string) (o *Observable) {
	//new Observable
	o = newObservable()
	o.Name = name

	//chain Observables
	parent.next = o
	o.pred = parent
	o.root = parent.root

	//set options
	o.buf_len = BufferLen
	return
}

// Debounce only emit an item from an Observable if a particular timespan has passed without it emitting another item
func (parent *Observable) Debounce(debounce time.Duration) (o *Observable) {
	o = parent.newFilterObservable("debounce")
	o.debounce = debounce
	o.distinct = false
	o.elementAt = 0
	o.first = false
	o.last = false
	o.sample = 0
	o.skip = 0
	o.skipLast = 0
	o.take = 0
	o.takeLast = 0
	o.operator = debounceOperator
	return o
}

var debounceOperator = filterOperator{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {
	var params = []reflect.Value{x}
	x = params[0]

	if !end {
		end = o.sendToFlow(ctx, x.Interface(), out)
	}
	return
}}

// Distinct suppress duplicate items emitted by an Observable
func (parent *Observable) Distinct() (o *Observable) {
	o = parent.newFilterObservable("distinct")
	o.debounce = 0
	o.distinct = true
	o.elementAt = 0
	o.first = false
	o.last = false
	o.sample = 0
	o.skip = 0
	o.skipLast = 0
	o.take = 0
	o.takeLast = 0
	o.operator = distinctOperator
	return o
}

var distinctOperator = filterOperator{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {
	var params = []reflect.Value{x}
	x = params[0]

	if !end {
		end = o.sendToFlow(ctx, x.Interface(), out)
	}
	return
}}

// ElementAt emit only item n emitted by an Observable
func (parent *Observable) ElementAt(num int) (o *Observable) {
	o = parent.newFilterObservable("elementAt")
	o.debounce = 0
	o.distinct = false
	o.elementAt = num
	o.first = false
	o.last = false
	o.sample = 0
	o.skip = 0
	o.skipLast = 0
	o.take = 0
	o.takeLast = 0
	o.operator = elementAtOperator
	return
}

var elementAtOperator = filterOperator{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {
	var params = []reflect.Value{x}
	x = params[0]

	if !end {
		end = o.sendToFlow(ctx, x.Interface(), out)
	}
	return
}}

// First emit only the first item (or the first item that meets some condition) emitted by an Observable
func (parent *Observable) First() (o *Observable) {
	o = parent.newFilterObservable("first")
	o.debounce = 0
	o.distinct = false
	o.elementAt = 0
	o.first = true
	o.last = false
	o.sample = 0
	o.skip = 0
	o.skipLast = 0
	o.take = 0
	o.takeLast = 0
	o.operator = firstOperator
	return o

}

var firstOperator = filterOperator{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {
	var params = []reflect.Value{x}
	x = params[0]

	if !end {
		end = o.sendToFlow(ctx, x.Interface(), out)
	}
	return
}}

// Last emit only the last item (or the last item that meets some condition) emitted by an Observable
func (parent *Observable) Last() (o *Observable) {
	o = parent.newFilterObservable("last")
	o.debounce = 0
	o.distinct = false
	o.elementAt = 0
	o.first = false
	o.last = true
	o.sample = 0
	o.skip = 0
	o.skipLast = 0
	o.take = 0
	o.takeLast = 0
	o.operator = lastOperator
	return o
}

var lastOperator = filterOperator{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {
	var params = []reflect.Value{x}
	x = params[0]

	if !end {
		end = o.sendToFlow(ctx, x.Interface(), out)
	}
	return
}}

// Sample emit the most recent items emitted by an Observable within periodic time intervals
func (parent *Observable) Sample(sample time.Duration) (o *Observable) {
	o = parent.newFilterObservable("sample")
	o.debounce = 0
	o.distinct = false
	o.elementAt = 0
	o.first = false
	o.last = false
	o.sample = sample
	o.skip = 0
	o.skipLast = 0
	o.take = 0
	o.takeLast = 0
	o.operator = sampleOperator
	return o
}

var sampleOperator = filterOperator{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {
	var params = []reflect.Value{x}
	x = params[0]

	if !end {
		end = o.sendToFlow(ctx, x.Interface(), out)
	}
	return
}}

// Skip suppress the first n items emitted by an Observable
func (parent *Observable) Skip(num int) (o *Observable) {
	o = parent.newFilterObservable("skip")
	o.debounce = 0
	o.distinct = false
	o.elementAt = 0
	o.first = false
	o.last = false
	o.sample = 0
	o.skip = num
	o.skipLast = 0
	o.take = 0
	o.takeLast = 0
	o.operator = skipOperator
	return o
}

var skipOperator = filterOperator{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {
	var params = []reflect.Value{x}
	x = params[0]

	if !end {
		end = o.sendToFlow(ctx, x.Interface(), out)
	}
	return
}}

// SkipLast suppress the final n items emitted by an Observable
func (parent *Observable) SkipLast(num int) (o *Observable) {
	o = parent.newFilterObservable("skipLast")
	o = parent.newFilterObservable("skip")
	o.debounce = 0
	o.distinct = false
	o.elementAt = 0
	o.first = false
	o.last = false
	o.sample = 0
	o.skip = 0
	o.skipLast = num
	o.take = 0
	o.takeLast = 0
	o.operator = skipLastOperator
	return o
}

var skipLastOperator = filterOperator{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {
	var params = []reflect.Value{x}
	x = params[0]

	if !end {
		end = o.sendToFlow(ctx, x.Interface(), out)
	}
	return
}}

// Take emit only the first n items emitted by an Observable
func (parent *Observable) Take(num int) (o *Observable) {
	o = parent.newFilterObservable("Take")
	o = parent.newFilterObservable("skip")
	o.debounce = 0
	o.distinct = false
	o.elementAt = 0
	o.first = false
	o.last = false
	o.sample = 0
	o.skip = 0
	o.skipLast = 0
	o.take = num
	o.takeLast = 0
	o.operator = takeOperator
	return o
}

var takeOperator = filterOperator{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {
	var params = []reflect.Value{x}
	x = params[0]

	if !end {
		end = o.sendToFlow(ctx, x.Interface(), out)
	}
	return
}}

// TakeLast emit only the final n items emitted by an Observable
func (parent *Observable) TakeLast(num int) (o *Observable) {
	o = parent.newFilterObservable("takeLast")
	o = parent.newFilterObservable("skip")
	o.debounce = 0
	o.distinct = false
	o.elementAt = 0
	o.first = false
	o.last = false
	o.sample = 0
	o.skip = 0
	o.skipLast = 0
	o.take = 0
	o.takeLast = num
	o.operator = takeLastOperator
	return o
}

var takeLastOperator = filterOperator{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {
	var params = []reflect.Value{x}
	x = params[0]

	if !end {
		end = o.sendToFlow(ctx, x.Interface(), out)
	}
	return
}}
